package com.culture.bootdemo.controller.thread;

import com.alibaba.fastjson.JSON;
import com.culture.bootdemo.dao.AnswerMapper;
import com.culture.bootdemo.po.Answer;
import com.google.common.collect.Lists;
import com.google.common.collect.Maps;
import com.google.common.util.concurrent.ThreadFactoryBuilder;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.collections4.ListUtils;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;

import java.util.*;
import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.atomic.AtomicLong;
import java.util.concurrent.atomic.LongAdder;

/**
 * @author: wen
 * @date 2020/10/27
 * 多线程
 */
@RestController
@RequestMapping("/thread")
@Slf4j
public class ThreadController {
    /**
     * 线程池创建线程-无界队列(LinkedBlockingQueue)   ThreadFactoryBuilder + ThreadPoolExecutor
     */
    private static final ExecutorService EXECUTOR = new ThreadPoolExecutor(8, 9, 0L, TimeUnit.MILLISECONDS, new LinkedBlockingQueue<>(), new ThreadFactoryBuilder().setNameFormat("demo-pool-%d").build());

    /**
     * 有界队列(ArrayBlockingQueue)   ThreadFactoryBuilder + ThreadPoolExecutor
     */
    private static final ExecutorService EXECUTOR1 = new ThreadPoolExecutor(8, 9, 0L, TimeUnit.MILLISECONDS, new ArrayBlockingQueue<>(10), new ThreadFactoryBuilder().setNameFormat("test-pool").build());

    @Autowired
    private AnswerMapper answerMapper;

    /**
     * Executors 创建线程
     */
    @GetMapping("test1")
    public void test1() {
        //创建单个线程
        Executors.newSingleThreadExecutor().execute(() -> System.out.println(11));
        //创建指定线程数- 无界队列，所以最大线程数无效
        ExecutorService fixedThreadPool = Executors.newFixedThreadPool(10);
        //阻塞队列 SynchronousQueue-不存储任务，只对任务进行中转 每新来一个任务，都会新建一个线程取执行
        ExecutorService cachedThreadPool = Executors.newCachedThreadPool();
        //延时队列 DelayedWorkQueue  延时队列可将任务按时间排序
        ScheduledExecutorService scheduledThreadPool = Executors.newScheduledThreadPool(4, new ThreadFactoryBuilder().setNameFormat("scheduledThreadPool-pool-%d").build());

        //线程池
        EXECUTOR.execute(() -> System.out.println(22));
        EXECUTOR.execute(() -> System.out.println(Thread.currentThread().getName()));
    }

    /**
     * CompletableFuture supplyAsync方法有返回结果值，可阻塞主线程
     */
    @GetMapping("test2-1")
    public void test2() {
        Integer integer = null;
        CompletableFuture<Integer> future = CompletableFuture.supplyAsync(() -> {
            return 24;
        }, EXECUTOR);
        try {
            integer = future.get();
        } catch (Exception e) {
            e.printStackTrace();
        }
        log.info("future返回数据:{}", integer);
        log.info("走到这里了111111");

        //==============保证多个线程执行完，再执行主线程下面的代码
        log.info("=====================================================================");

        List<CompletableFuture<Integer>> listFuture = Collections.synchronizedList(new ArrayList<>());
        List<Integer> result = new ArrayList<>();
        List<Integer> initList = Arrays.asList(1, 2, 3, 4, 5, 6, 7, 8, 9, 10);
        for (Integer i : initList) {
            CompletableFuture<Integer> futureSingle = CompletableFuture.supplyAsync(() -> {
                //手动写异常
                if (i == 3) {
                    int a = 1 / 0;
                }
                log.info("当前线程名称:{}对应数据==>{}", Thread.currentThread().getName(), i);
                return i;
            }, EXECUTOR);
            listFuture.add(futureSingle);
        }

        // get方法 为了阻塞主线程，从而不使多线程任务中断
        listFuture.forEach(a -> {
            try {
                result.add(a.get());
            } catch (Exception e) {
                e.printStackTrace();
            }
        });
        log.info("拼装数据:{}", JSON.toJSONString(result));
        log.info("走到这里了222222222");
    }

    /**
     * 大数据量情况下：分批查询数据===方法一：
     * CompletableFuture
     * supplyAsync 异步有返回值
     * get方法阻塞
     */
    @GetMapping("test2-1-1")
    public void test221() {

        //数据总量,应先从数据库中查询出总量
        int count = 10;
        //每次查询数量
        int limit = 3;
        // 遍历循环的次数
        int pages = count / limit + 1;
        List<CompletableFuture<List<Answer>>> listFuture = new ArrayList<>();
        List<Answer> resultList = Collections.synchronizedList(new ArrayList<>());
        for (int i = 0; i < pages; i++) {
            int offSet = i * limit;
            CompletableFuture<List<Answer>> completableFuture = CompletableFuture.supplyAsync(() -> {
                if (offSet == 6) {
                    try {
                        Thread.sleep(5000);
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                }
                return answerMapper.selectByLimit(offSet, limit);
            }, EXECUTOR);
            listFuture.add(completableFuture);
        }
        for (CompletableFuture<List<Answer>> cf : listFuture) {
            try {
                resultList.addAll(cf.get());
            } catch (InterruptedException | ExecutionException e) {
                log.error("test221 error", e);
                throw new RuntimeException(e);
            }
        }
        log.info("list:{}", JSON.toJSONString(resultList));

    }


    /**
     * 大数据量情况下：分批查询数据===方法二：
     * CompletableFuture
     * supplyAsync 异步有返回值
     * allOf :当所有给定的 CompletableFuture 完成时，返回一个新的 CompletableFuture，必须配合join使用，才会阻塞
     * whenComplete:对每个CompletableFuture返回结果的处理 (如果出现异常，则整个会抛异常)
     */
    @GetMapping("test2-1-2")
    public void test212() {

        //数据总量,应先从数据库中查询出总量
        int count = 10;
        //每次查询数量
        int limit = 3;
        // 遍历循环的次数
        int pages = count / limit + 1;
        List<CompletableFuture<List<Answer>>> listFuture = new ArrayList<>();
        List<Answer> resultList = Collections.synchronizedList(new ArrayList<>());
        for (int i = 0; i < pages; i++) {
            int offSet = i * limit;
            CompletableFuture<List<Answer>> completableFuture = CompletableFuture.supplyAsync(() -> {
                if (offSet == 6) {
                    try {
                        Thread.sleep(5000);
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                }
                return answerMapper.selectByLimit(offSet, limit);
            }, EXECUTOR).whenComplete((r, e) -> {
                resultList.addAll(r);
            });
            listFuture.add(completableFuture);
        }
        CompletableFuture.allOf(listFuture.toArray(new CompletableFuture[0])).join();
        log.info("test2-1-2-list:{}", JSON.toJSONString(resultList));

    }


    /**
     * thenAccept()处理正常结果；
     * exceptional()处理异常结果；
     * <p>
     * handle:在任务完成后再执行，还可以处理异常的任务（不会抛异常，会部分成功）
     */
    @GetMapping("test2-1-3")
    public void test213() {

        //数据总量,应先从数据库中查询出总量
        int count = 10;
        //每次查询数量
        int limit = 3;
        // 遍历循环的次数
        int pages = count / limit + 1;
        List<CompletableFuture<List<Answer>>> listFuture = new ArrayList<>();
        List<Answer> resultList = Collections.synchronizedList(new ArrayList<>());
        for (int i = 0; i < pages; i++) {
            int offSet = i * limit;
            CompletableFuture<List<Answer>> completableFuture = CompletableFuture.supplyAsync(() -> {
                if (offSet == 6) {
                    int v = 1 / 0;
                }
                return answerMapper.selectByLimit(offSet, limit);
            }, EXECUTOR).handle((r, e) -> {
                if (null != e) {
                    log.error("error----:{}；r:{}", e.getMessage(),JSON.toJSONString(r));
                } else {
                    resultList.addAll(r);
                }
                return r;
            });

            listFuture.add(completableFuture);
        }
        CompletableFuture.allOf(listFuture.toArray(new CompletableFuture[0])).join();
        log.info("test2-1-2-list:{}", JSON.toJSONString(resultList));

    }


    /**
     * CompletableFuture  supplyAsync返回值
     * whenComplete 阻塞CompletableFuture线程并返回结果值
     * join() 阻塞主线程，即：CompletableFuture有结果值后再进行主线程代码，否则不加join()的话，会视CompletableFuture为异步，直接走完主线程
     */
    @GetMapping("test2-2")
    public void test23() {
        Integer integer = 2;
        CompletableFuture.supplyAsync(() -> {
            try {
                Thread.sleep(5000);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            return integer * 3;
        }, EXECUTOR).whenComplete((result, e) -> {
            //whenComplete 阻塞CompletableFuture线程并返回结果值
            if (null != e) {
                log.error("报错了", e);
            } else {
                log.info("返回值是：{}", result);
            }
        }).join();
        log.info("走到这里了111111");

    }

    /**
     * CompletableFuture
     */
    @GetMapping("test2-3")
    public void test3() {
        List<CompletableFuture<List<String>>> listFuture = new ArrayList<>();
        List<String> results = Collections.synchronizedList(new ArrayList<>());
        CompletableFuture<List<String>> future = CompletableFuture.supplyAsync(() -> {
            List<String> list = new ArrayList<>();
            //业务代码
            for (int i = 0; i < 10; i++) {
                list.add(String.valueOf(i));
                System.out.println(Thread.currentThread().getName() + "---" + i);
            }
            results.addAll(list);
            return list;
        }, EXECUTOR);

        listFuture.add(future);

        // get方法 为了阻塞主线程，从而不使多线程任务中断
        listFuture.forEach(a -> {
            try {
                a.get();
            } catch (Exception e) {
                e.printStackTrace();
            }
        });

        System.out.println(results);
    }

    /**
     * CompletableFuture 设置超时时间: get方法(需手动捕获异常)
     * sleep 线程休眠时间
     */
    @GetMapping("test2-4")
    public void test7() {
        try {
            CompletableFuture.runAsync(() -> {
                try {
                    Thread.sleep(5000);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }

                try {
                    TimeUnit.SECONDS.sleep(5);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }).get(1, TimeUnit.MINUTES);
        } catch (Exception e) {
            e.printStackTrace();
        }
        log.info("33333333333333");
    }


    /**
     * CompletableFuture join方法(不会强制抛异常)
     */
    @GetMapping("test2-5")
    public void test25() {
        List<Integer> list = new ArrayList<>(Arrays.asList(1, 2, 3, 4, 5, 6, 7, 8, 9));
        List<List<Integer>> partition = ListUtils.partition(list, 3);
        System.out.println(partition);

        partition.stream().forEach(p -> {
            CompletableFuture.supplyAsync(() -> {
                if (p.contains(3)) {
                    try {
                        Thread.sleep(5000);
                        System.out.println("pppppp:" + p);
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                }
                return 0;
            }, EXECUTOR).join();
        });

        System.out.println(1111111);
    }


    /**
     * FutureTask: 有结果值，可阻塞主线程
     */
    @GetMapping("test4")
    public void test4() {
        FutureTask<String> task = new FutureTask<>(() -> {
            //业务代码
            return "哈喽";
        });
        EXECUTOR.execute(task);
        boolean done = task.isDone();
        long l1 = System.currentTimeMillis();
        //这个方法会阻塞，直到任务完成时会返回
        String s = null;

        try {
            s = task.get();
        } catch (Exception e) {
            e.printStackTrace();
        }

        boolean done1 = task.isDone();
        long l2 = System.currentTimeMillis();
        System.out.println("任务返回结果 = " + done + "   " + done1 + "    time = " + (l2 - l1) + "   " + s);

        //============================================================================//


        List<String> results = Lists.newArrayList();
        List<FutureTask<String>> futureTasks = Lists.newArrayList();
        futureTasks.add(new FutureTask<>(() -> {
            //业务代码
            return "sss";
        }));
        futureTasks.add(new FutureTask<>(() -> {
            //业务代码
            return "www";
        }));
        futureTasks.add(new FutureTask<>(() -> {
            //业务代码
            return "eee";
        }));
        for (FutureTask ft : futureTasks) {
            EXECUTOR.execute(ft);
        }
        for (FutureTask<String> futureTask : futureTasks) {
            try {
                //起阻塞作用，也可以获取返回值
                //Java获取结果的话，可以利用值传递获取
                results.add(futureTask.get());

            } catch (Exception e) {
                e.printStackTrace();
            }
        }
        System.out.println(results);
    }

    /**
     * FutureTask: 有结果值，可阻塞主线程
     */
    @GetMapping("test5")
    public void test5() {
        List<String> strings = Lists.newArrayList("1", "2", "3", "4", "5", "6", "7", "8", "9", "10", "11", "12");
        List<String> results = Lists.newArrayList();
        List<FutureTask<List<String>>> futureTasks = Lists.newArrayList();
        futureTasks.add(new FutureTask<>(() -> {
            List<String> list = new ArrayList<>();
            //业务代码
            for (String s : strings) {
                list.add(String.valueOf(s));
            }
            return list;
        }));
        for (FutureTask<List<String>> ft : futureTasks) {
            EXECUTOR.execute(ft);
        }
        for (FutureTask<List<String>> futureTask : futureTasks) {
            try {
                //起阻塞作用，也可以获取返回值
                //Java获取结果的话，可以利用值传递获取
                results.addAll(futureTask.get());
                //方式二：采用值传递方式获取list，此处代码只有：{futureTask.get()},但results应设置为线程安全的类

            } catch (Exception e) {
                e.printStackTrace();
            }
        }
        System.out.println(results);
    }


    private static final int limit = 8;
    private static final ExecutorService execCountService = Executors.newFixedThreadPool(limit);

    /**
     * 多线程处理大list，将list拆分，并使得每个线程处理一个list
     * 指定8个线程处理  :
     * CountDownLatch  : 减法计数器，计算为0时释放所有等待的线程
     * AtomicInteger
     * LongAdder
     */
    @GetMapping("test6")
    public void test6() {
        List<String> strings = Lists.newArrayList("1", "2", "3", "4", "5", "6", "7", "8", "9");
        List<List<String>> partition = ListUtils.partition(strings, strings.size() / limit);
        //初始化 CountDownLatch
        CountDownLatch cdl = new CountDownLatch(partition.size());
        log.info("将list按照大小为:{}进行拆分", (strings.size() / limit));
        log.info("partion大小(共拆分成多少小list):{}", partition.size());
        AtomicInteger failCount = new AtomicInteger();

        //GUC下原子类 同AtomicInteger 高并发下 用此类
        LongAdder failCount2 = new LongAdder();
        try {
            for (List<String> list : partition) {
                execCountService.execute(() -> {
                    try {
                        if ("3".equals(list.get(0))) {
                            int a = 1 / 0;
                        }
                        log.info("当前线程名称:{}==>{}", Thread.currentThread().getName(), JSON.toJSONString(list));
                    } catch (Exception e) {
                        failCount.incrementAndGet();
                        failCount2.increment();
                        log.error("多线程错误", e);
                        log.error("计算错误:{}==>{}", Thread.currentThread().getName(), JSON.toJSONString(list));
                    } finally {
                        cdl.countDown();
                        log.info("计数器是：{}", cdl.getCount());
                    }
                });
            }
        } catch (Exception e) {
            log.info("test5解析报错:", e);
        }

        try {
            cdl.await();
        } catch (InterruptedException e) {
            log.info("countDownLatch错误", e);
        }

        //失败总次数
        int failSum = failCount2.intValue();
        log.info("走到这里了吗");
    }

    /**
     * countDownLatch的倒用法：
     * 一般用法：是主线程等待其他线程都执行完毕后，主线程再开始进行下面的操作
     * 本实例采用倒用方法：多个线程等待某一个线程的信号，同时开始执行
     */
    @GetMapping("countDownLatchTest")
    public void testCountDownLatch() {
        // 1.创建一个5的线程池
        // 2.新建CountDownLatch，倒数为1
        // 3.for循环方式提交5个任务，5个任务一开始就执行await()方法进行等待
        // 4.接着回到主线程，当主线程休息5s后，countDown减1，一旦主线程调用了countDown()方法，则那5个已经调用了await()方法的线程被唤醒，继续执行await()方法后面的操作
        log.info("运动员有5秒的准备时间");
        CountDownLatch countDownLatch = new CountDownLatch(1);
        ExecutorService service = Executors.newFixedThreadPool(5);
        for (int i = 0; i < 5; i++) {
            final int no = i + 1;
            Runnable runnable = () -> {
                log.info(no + "号运动员准备完毕，等待裁判员的发令枪");
                try {
                    countDownLatch.await();
                    log.info(no + "号运动员开始跑步了");
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            };
            service.submit(runnable);
        }
        try {
            Thread.sleep(5000);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        log.info("5秒准备时间已过，发令枪响，比赛开始！");
        countDownLatch.countDown();
    }

    /**
     * CyclicBarrier: N个线程相互等待，任何一个线程完成之前，所有的线程都必须等待
     * 加法计数器：计数达到指定值时释放所有等待线程；计数达到指定值时，计数置为0重新开始，可重复利用
     */
    @GetMapping("test8")
    public void test8() {
        CyclicBarrier cyclicBarrier = new CyclicBarrier(4);

        for (int i = 0; i < 10; i++) {
            EXECUTOR.execute(() -> {
                try {
                    log.info("当前线程名称:{}", Thread.currentThread().getName());
                    cyclicBarrier.await();
                    log.info("cyclicBarrier初始为4，等待有四个线程到达后开始工作:");
                } catch (Exception e) {
                    //异常发生时，对失败计数器+1
                    log.info("Error : ", e);

                }
            });
        }

        log.info("结束==================");
    }


    /**
     * Java创建线程的方式
     *
     * @throws Exception
     */
    @GetMapping("createThread")
    public void createThread1() throws Exception {
        AtomicLong atomicLong = new AtomicLong(0);
        AtomicInteger atomicInteger = new AtomicInteger(0);

        //方式一：继承Thread
        for (int i = 0; i < 100; i++) {
            new Thread1(atomicInteger).start();
        }

        TimeUnit.SECONDS.sleep(2);
        System.out.println("thread1-Thread:" + atomicInteger.get());

        //方式二：实现Runnable接口
        for (int i = 0; i < 100; i++) {
            new Thread(new Thread2(atomicLong)).start();
        }
        TimeUnit.SECONDS.sleep(2);
        System.out.println("thread2-Runnable:" + atomicLong.get());

        //=============================================
        //实现Callable接口 结合Future ===========本身不创建线程，还是要通过Thread等方式创建线程
        ExecutorService pool = Executors.newFixedThreadPool(10);
        Future<Integer> future = pool.submit(new CallableTask());
        try {
            System.out.println(future.get());
        } catch (InterruptedException | ExecutionException e) {
            e.printStackTrace();
        }
        pool.shutdown();

        //Callable 结合 FutureTask实现==
        CallableTask task = new CallableTask();
        FutureTask<Integer> integerFutureTask = new FutureTask<>(task);
        new Thread(integerFutureTask).start();
//        Future<Integer> submit = EXECUTOR.submit(task);
        try {
            System.out.println("task运行结果：" + integerFutureTask.get());
        } catch (InterruptedException | ExecutionException e) {
            e.printStackTrace();
        }

    }


    //继承Thread类的方式创建
    static class Thread1 extends Thread {
        private final AtomicInteger integerCount;

        public Thread1(AtomicInteger counter) {
            this.integerCount = counter;
        }

        @Override
        public void run() {
            integerCount.incrementAndGet();
        }
    }

    //实现Runnable接口方式创建 >>> 更好
    static class Thread2 implements Runnable {
        private final AtomicLong longCount;

        public Thread2(AtomicLong counter) {
            this.longCount = counter;
        }

        @Override
        public void run() {
            longCount.incrementAndGet();
        }
    }

    //Callable： 有返回值，能抛出异常
    static class CallableTask implements Callable<Integer> {
        @Override
        public Integer call() throws Exception {
            Thread.sleep(3000);
            return new Random().nextInt();
        }
    }


}



